package com.yk.queue;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

/**
 * @Date: 2018/11/12 17:39
 * @Version: 1.0
 * @Author: pengqingfeng
 * @Description:
 */
@Component
public class KafkaConsumer {
    protected final Logger log = LoggerFactory.getLogger(this.getClass());

//    @Async("executorPool")
//    @KafkaListener(topics = {"test"})
    public void receiveMessage(ConsumerRecord<?, ?> record) {
        get(record, "test-------");
    }

    @KafkaListener(topics = {"test1"})
    public void receiveMessage1(ConsumerRecord<?, ?> record) {
        get(record, "test1-------");
    }


////    @Async("executorPool")
//    @KafkaListener(topics = {"test"})
//    public void receiveMessage1(ConsumerRecord<?, ?> record) {
//        get(record, "bbbbbb-------");
//    }
////    @Async("executorPool")
//    @KafkaListener(topics = {"test"})
//    public void receiveMessage2(ConsumerRecord<?, ?> record) {
//        get(record, "cccccc-------");
//    }


    public void get(ConsumerRecord<?, ?> record, String str) {
        try {
            Thread.sleep(1);
            log.info(str + "接收消息为,partition:{}----key:{}---value:{}", record.partition(), record.key(), record.value());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
